// Copyright (c) The Libra Core Contributors
// SPDX-License-Identifier: Apache-2.0

mod block_processor;
mod block_tree;
mod transaction_block;

#[cfg(test)]
mod executor_test;
#[cfg(test)]
mod mock_vm;

use crate::block_processor::BlockProcessor;
use config::config::NodeConfig;
use crypto::{
    hash::{GENESIS_BLOCK_ID, PRE_GENESIS_BLOCK_ID, SPARSE_MERKLE_PLACEHOLDER_HASH},
    HashValue,
};
use execution_proto::{CommitBlockResponse, ExecuteBlockResponse, ExecuteChunkResponse};
use failure::{format_err, Result};
use futures::{channel::oneshot, executor::block_on};
use lazy_static::lazy_static;
use logger::prelude::*;
use std::{
    collections::HashMap,
    marker::PhantomData,
    sync::{mpsc, Arc, Mutex},
};
use storage_client::{StorageRead, StorageWrite};
use types::{
    ledger_info::{LedgerInfo, LedgerInfoWithSignatures},
    transaction::{SignedTransaction, TransactionListWithProof},
};
use vm_runtime::VMExecutor;

lazy_static! {
    static ref OP_COUNTERS: metrics::OpMetrics = metrics::OpMetrics::new_and_registered("executor");
}

/// `Executor` implements all functionalities the execution module needs to provide.
pub struct Executor<V> {
    /// A thread that keeps processing blocks.
    block_processor_thread: Option<std::thread::JoinHandle<()>>,

    /// Where we can send command to the block processor. The block processor sits at the other end
    /// of the channel and processes the commands.
    command_sender: Mutex<Option<mpsc::Sender<Command>>>,

    phantom: PhantomData<V>,
}

impl<V> Executor<V>
where
    V: VMExecutor,
{
    /// Constructs an `Executor`.
    pub fn new(
        storage_read_client: Arc<dyn StorageRead>,
        storage_write_client: Arc<dyn StorageWrite>,
        config: &NodeConfig,
    ) -> Self {
        let startup_info = storage_read_client
            .get_executor_startup_info()
            .expect("Failed to read startup info from storage.");

        let (
            state_root_hash,
            frozen_subtrees_in_accumulator,
            num_elements_in_accumulator,
            committed_timestamp_usecs,
            committed_block_id,
        ) = match startup_info {
            Some(info) => {
                info!("Startup info read from DB: {:?}.", info);
                let ledger_info = info.ledger_info;
                (
                    info.account_state_root_hash,
                    info.ledger_frozen_subtree_hashes,
                    ledger_info.version() + 1,
                    ledger_info.timestamp_usecs(),
                    ledger_info.consensus_block_id(),
                )
            }
            None => {
                info!("Startup info is empty. Will start from GENESIS.");
                (
                    *SPARSE_MERKLE_PLACEHOLDER_HASH,
                    vec![],
                    0,
                    0,
                    *PRE_GENESIS_BLOCK_ID,
                )
            }
        };

        let (command_sender, command_receiver) = mpsc::channel();

        let vm_config = config.vm_config.clone();
        let executor = Executor {
            block_processor_thread: Some(
                std::thread::Builder::new()
                    .name("block_processor".into())
                    .spawn(move || {
                        let mut block_processor = BlockProcessor::<V>::new(
                            command_receiver,
                            committed_timestamp_usecs,
                            state_root_hash,
                            frozen_subtrees_in_accumulator,
                            num_elements_in_accumulator,
                            committed_block_id,
                            storage_read_client,
                            storage_write_client,
                            vm_config,
                        );
                        block_processor.run();
                    })
                    .expect("Failed to create block processor thread."),
            ),
            command_sender: Mutex::new(Some(command_sender)),
            phantom: PhantomData,
        };

        if num_elements_in_accumulator == 0 {
            let genesis_transaction = config
                .execution
                .get_genesis_transaction()
                .expect("failed to load genesis transaction!");
            executor.init_genesis(genesis_transaction);
        }

        executor
    }

    /// This is used when we start for the first time and the DB is completely empty. It will write
    /// necessary information to DB by committing the genesis transaction.
    fn init_genesis(&self, genesis_txn: SignedTransaction) {
        // Create a block with genesis_txn being the only transaction. Execute it then commit it
        // immediately.
        // We create `PRE_GENESIS_BLOCK_ID` as the parent of the genesis block.
        let response = block_on(self.execute_block(
            vec![genesis_txn],
            *PRE_GENESIS_BLOCK_ID,
            *GENESIS_BLOCK_ID,
        ))
        .expect("Response sender was unexpectedly dropped.")
        .expect("Failed to execute genesis block.");

        let root_hash = response.root_hash();
        let ledger_info = LedgerInfo::new(
            /* version = */ 0,
            root_hash,
            /* consensus_data_hash = */ HashValue::zero(),
            *GENESIS_BLOCK_ID,
            /* epoch_num = */ 0,
            /* timestamp_usecs = */ 0,
        );
        let ledger_info_with_sigs =
            LedgerInfoWithSignatures::new(ledger_info, /* signatures = */ HashMap::new());
        block_on(self.commit_block(ledger_info_with_sigs))
            .expect("Response sender was unexpectedly dropped.")
            .expect("Failed to commit genesis block.");
        info!("GENESIS transaction is committed.")
    }

    /// Executes a block.
    pub fn execute_block(
        &self,
        transactions: Vec<SignedTransaction>,
        parent_id: HashValue,
        id: HashValue,
    ) -> oneshot::Receiver<Result<ExecuteBlockResponse>> {
        debug!(
            "Received request to execute block. Parent id: {:x}. Id: {:x}.",
            parent_id, id
        );

        let (resp_sender, resp_receiver) = oneshot::channel();
        match self
            .command_sender
            .lock()
            .expect("Failed to lock mutex.")
            .as_ref()
        {
            Some(sender) => sender
                .send(Command::ExecuteBlock {
                    transactions,
                    parent_id,
                    id,
                    resp_sender,
                })
                .expect("Did block processor thread panic?"),
            None => resp_sender
                .send(Err(format_err!("Executor is shutting down.")))
                .expect("Failed to send error message."),
        }
        resp_receiver
    }

    /// Commits a block and all its ancestors.
    pub fn commit_block(
        &self,
        ledger_info_with_sigs: LedgerInfoWithSignatures,
    ) -> oneshot::Receiver<Result<CommitBlockResponse>> {
        debug!(
            "Received request to commit block {:x}.",
            ledger_info_with_sigs.ledger_info().consensus_block_id()
        );

        let (resp_sender, resp_receiver) = oneshot::channel();
        match self
            .command_sender
            .lock()
            .expect("Failed to lock mutex.")
            .as_ref()
        {
            Some(sender) => sender
                .send(Command::CommitBlock {
                    ledger_info_with_sigs,
                    resp_sender,
                })
                .expect("Did block processor thread panic?"),
            None => resp_sender
                .send(Err(format_err!("Executor is shutting down.")))
                .expect("Failed to send error message."),
        }
        resp_receiver
    }

    /// Executes and commits a chunk of transactions that are already committed by majority of the
    /// validators.
    pub fn execute_chunk(
        &self,
        txn_list_with_proof: TransactionListWithProof,
        ledger_info_with_sigs: LedgerInfoWithSignatures,
    ) -> oneshot::Receiver<Result<ExecuteChunkResponse>> {
        debug!(
            "Received request to execute chunk. Chunk size: {}. Target version: {}.",
            txn_list_with_proof.transaction_and_infos.len(),
            ledger_info_with_sigs.ledger_info().version(),
        );

        let (resp_sender, resp_receiver) = oneshot::channel();
        match self
            .command_sender
            .lock()
            .expect("Failed to lock mutex.")
            .as_ref()
        {
            Some(sender) => sender
                .send(Command::ExecuteChunk {
                    txn_list_with_proof,
                    ledger_info_with_sigs,
                    resp_sender,
                })
                .expect("Did block processor thread panic?"),
            None => resp_sender
                .send(Err(format_err!("Executor is shutting down.")))
                .expect("Failed to send error message."),
        }
        resp_receiver
    }
}

impl<V> Drop for Executor<V> {
    fn drop(&mut self) {
        // Drop the sender so the block processor thread will exit.
        self.command_sender
            .lock()
            .expect("Failed to lock mutex.")
            .take()
            .expect("Command sender should exist.");
        self.block_processor_thread
            .take()
            .expect("Block processor thread should exist.")
            .join()
            .expect("Did block processor thread panic?");
    }
}

#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
enum Command {
    ExecuteBlock {
        transactions: Vec<SignedTransaction>,
        parent_id: HashValue,
        id: HashValue,
        resp_sender: oneshot::Sender<Result<ExecuteBlockResponse>>,
    },
    CommitBlock {
        ledger_info_with_sigs: LedgerInfoWithSignatures,
        resp_sender: oneshot::Sender<Result<CommitBlockResponse>>,
    },
    ExecuteChunk {
        txn_list_with_proof: TransactionListWithProof,
        ledger_info_with_sigs: LedgerInfoWithSignatures,
        resp_sender: oneshot::Sender<Result<ExecuteChunkResponse>>,
    },
}
